//
// Created by root on 11/7/18.
//

#include <cstdlib>
#include <inttypes.h>
#include <string.h>
#include <errno.h>
#include <sys/uio.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/un.h>
#include <sys/sendfile.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/types.h>

//#include <st.h>
#include "ursax.h"
#include "log.h"
#include "anysocket.h"
#include "protocol.h"
#include "networking.h"
#include "utcache.h"
#include "debug.h"
#include "st-hack.h"
#include "mmap-manager.h"
//#include "mmap-manager.h"

#if EAGAIN != EWOULDBLOCK
#define _IO_NOT_READY_ERROR  ((errno == EAGAIN) || (errno == EWOULDBLOCK))
#else
#define _IO_NOT_READY_ERROR  (errno == EAGAIN)
#endif

bool show_socket_error = true;
bool multy_send = false;
uint32_t splice_line = 64*1024;

int parse_ip(char *p, uint32_t* out_ip)
{
    // LOG_DEBUG(p);
    uint32_t ip = ntohl(inet_addr(p));
    if (unlikely(ip == INADDR_NONE))
    {
        LOG_ERROR("invalid ip address: '%s'", p);
        return -1;
    }
    *out_ip = ip;
    return 0;
}

int parse_port(char *p, uint16_t* out_port)
{
    uint32_t port;
    int ret = sscanf(p, "%u", &port);
    if (unlikely(ret != 1 || port > 0xffff))
    {
        LOG_ERROR("invalid TCP/UDP port: '%s'", p);
        return -1;
    }
    *out_port = (uint16_t)port;
    return 0;
}

// 0 for success, -1 for failed
int parse_ip_port(char* s, uint32_t* ip, uint16_t* port, uint16_t default_port)
{
    char* p = strchr(s, ':');
    if (p)
    {
        int a = parse_port(p+1, port);
        char c = *p;
        *p = '\0';
        int b = parse_ip(s, ip);
        *p = c;
        return a | b;	// both success (0) for sucess
    }
    else
    {
        *port = default_port;
        return parse_ip(s, ip);
    }
}

st_netfd_t ursax_connect(uint32_t ip, uint16_t port, bool do_retry)
{
    LOG_DEBUG("(%s:%d) enter", str_ip(ip), port);
    if(unlikely(ip == 0)){
        debug_reference(0);
    }

    int osfd = socket(PF_INET, SOCK_STREAM, 0);
    if (osfd < 0)
    {
        LOG_ERROR("failed to create new socket %d: %d (%s:%d)", errno, strerror(errno), str_ip(ip), port);
        return NULL;
    }

    st_netfd_t connection = st_netfd_open_socket(osfd);
    enable_reuse_addr(osfd);

    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = htonl(ip);

    int ret, retry = 0;
    again:
    ret = st_connect(connection, (struct sockaddr*)&addr, sizeof(addr), 1*1000*1000);
    if (unlikely(ret < 0))
    {
        if (errno == EADDRINUSE) // todo EADDRINUSE ?
        {
            if (likely(foo_cache_release_any() > 0))
                goto again;
        }
        LOG_ERROR("unable to connect to remote host %s:%u, %d: %s", str_ip(ip), port, errno, strerror(errno));
        if (retry < 6 && do_retry)
        {
            st_usleep((1<<retry) * 100 * 1000);
            retry++;
            goto again;
        }
        st_netfd_close(connection);
        return NULL;
    }

    enable_tcp_no_delay(osfd);
    enable_tcp_cork(osfd);
    set_tcp_wbuf_size(osfd);
    set_tcp_rbuf_size(osfd);
    return connection;
}

int getsockname_ex(int socket, char* buf)
{
    struct sockaddr_in addr;
    socklen_t len = sizeof(addr);
    int ret = getsockname(socket, (struct sockaddr*)&addr, &len);
    if (unlikely(ret < 0))
    {
        LOG_ERROR("failed to getsockname %d: %s", errno, strerror(errno));
        return ret;
    }

    strcpy(buf, inet_ntoa(addr.sin_addr));
    return 0;
}

int gethostbypeer(uint32_t ip, char* hostname)
{
    int s = socket(PF_INET, SOCK_DGRAM, 0);
    if (unlikely(s < 0)) {
        LOG_ERROR("failed to create UDP socket %d: %s", errno, strerror(errno));
        return s;
    }

    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(12345);
    addr.sin_addr.s_addr = htonl(ip);
    int ret = connect(s, (struct sockaddr*)&addr, sizeof(addr));
    if (unlikely(ret < 0)) {
        close(s);
        LOG_ERROR("failed to connect UDP socket %d: %s", errno, strerror(errno));
        return ret;
    }

    ret = getsockname_ex(s, hostname);
    close(s);
    return 0;
}

int gethostbypeer_str(char* peer, char* hostname)
{
    uint32_t ip;
    int ret = parse_ip(peer, &ip);
    if (ret < 0)
        return ret;

    return gethostbypeer(ip, hostname);
}

int recv_drop_or_log(st_netfd_t stfd, size_t size, const char* logstr)
{
    int x = errno;
    while(size > 0) {
        char buf[4096];
        int len = size;
        if (unlikely(len > (int)sizeof(buf)-1)) {
            len = sizeof(buf) -1;
        }
        len = st_read_fully(stfd, buf, len, URSAX_TIMEOUT);
        if (unlikely(len <= 0)) {
            LOG_INFO("message recv failed %d: %s", errno, strerror(errno));
            return len;
        }
        if (logstr) {
            buf[len] = 0;
            LOG_INFO("%s: %s", logstr, buf);
        }
        size -= len;
    }
    errno = x;
    return 0;
}

// this is in dll, avoid conflict from operation_send_recv in nbd, for the hack of st_write st_read_fully
int operation_send_recv_tmout(st_netfd_t stfd, void* req,
                              uint32_t req_size, void* resp,
                              uint32_t resp_size, uint32_t w_tmout, uint32_t r_tmout)
{
    LOG_DEBUG("%s(fd=%u, req=(%p, %u), resp=(%p, %u))", __func__, stfd, req, req_size, resp, resp_size);

    int len = st_write(stfd, req, req_size, (st_utime_t)w_tmout);
    if (unlikely(len < (int)req_size))
    {
        int errno_st_write = errno;
        log_socket_error(stfd, "write");
        return -errno_st_write;
    }
    flush_tcp(stfd);

    len = st_read_fully(stfd, resp, resp_size, (st_utime_t)r_tmout);
    if (unlikely(len < (int)resp_size))
    {
        int errno_st_read_fully = errno;
        log_socket_error(stfd, "read");
        return -errno_st_read_fully;
    }

    return len;
}

int operation_send_recv_with_timeout(st_netfd_t stfd, void* req,
        uint32_t req_size, void* resp, uint32_t resp_size, uint32_t utimeout)
{
    return operation_send_recv_tmout(stfd, req, req_size, resp, resp_size,
            utimeout, utimeout);
}

int operation_send_recv(st_netfd_t stfd, void* req,
                        uint32_t req_size, void* resp, uint32_t resp_size)
{
    return operation_send_recv_tmout(stfd, req, req_size, resp, resp_size,
            URSAX_TIMEOUT, URSAX_TIMEOUT);
}

void log_socket_error(st_netfd_t stfd, const char* part)
{
    struct sockaddr_in remote_addr;
    socklen_t addrlen = sizeof(remote_addr);
    int _errno = errno;
    getpeername(st_netfd_fileno(stfd), (struct sockaddr*)&remote_addr, &addrlen);
    errno = _errno;
    if(show_socket_error) {
        LOG_ERROR("socket to (%s:%u) %s error %d: %s",
                  inet_ntoa(remote_addr.sin_addr), ntohs(remote_addr.sin_port),
                  part, errno, strerror(errno));
    }
    else {
        LOG_DEBUG("socket to (%s:%u) %s error %d: %s",
                  inet_ntoa(remote_addr.sin_addr), ntohs(remote_addr.sin_port),
                  part, errno, strerror(errno), print_call_stack());
    }
}

ssize_t st_sendfile(st_netfd_t out_fd, int fd, off_t *offset, size_t count, size_t* sends)
{
    ssize_t _count = count;
    while(count > 0)
    {
        ssize_t n = sendfile(st_netfd_fileno(out_fd), fd, offset, count);
        if (unlikely(n < 0)) {
            if (unlikely((errno == EINTR))) {
                LOG_INFO("sendfile errno=EINTR, continue");
                continue;
            }
            if (unlikely(!_IO_NOT_READY_ERROR)) {
                LOG_INFO("sendifle _IO_NOT_READY_ERROR %m");
                return -1;
            }
            /* Wait until the socket becomes writable */
            if (unlikely(st_netfd_poll(out_fd, POLLOUT, URSAX_TIMEOUT) < 0)) {
                LOG_INFO("sendfile st_netfd_poll wait for socket writable");
                return -1;
            }
            // LOG_INFO("sendfile continue, %m");
            continue;
        }
        else if(unlikely(n == 0)) {
            LOG_ERROR("sendfile fd sock %d fd file %d offset %d count %d %m", st_netfd_fileno(out_fd), fd, *offset, count);
            return -1;
        }
        count -= n;
        // *offset += n;  // offset自动增长
        if(sends) {
            (*sends)++;
        }
    }

    return _count;
}

int operation_send3_recv(st_netfd_t stfd,
                         void* part1, uint32_t part1_size,
                         void* part2, uint32_t part2_size,
                         void* part3, uint32_t part3_size,
                         void* resp, uint32_t resp_size)
{
    struct iovec iov[3];
    iov[0].iov_base = part1;
    iov[0].iov_len = part1_size;
    iov[1].iov_base = part2;
    iov[1].iov_len = part2_size;
    iov[2].iov_base = part3;
    iov[2].iov_len = part3_size;
    LOG_DEBUG("start write");
    int len = st_writev(stfd, iov, 3, URSAX_TIMEOUT);
    if(unlikely(len < (int)(part1_size + part2_size + part3_size))){
        int errno_st_writev = errno;
        log_socket_error(stfd, "writev");
        return -errno_st_writev;
    }
    flush_tcp(stfd);

    LOG_DEBUG("write over, start read");
    len = st_read_fully(stfd, resp, resp_size, URSAX_TIMEOUT);
    if(unlikely(len < (int)resp_size)){
        int errno_st_read_fully = errno;
        log_socket_error(stfd, "read resp");
        return -errno_st_read_fully;
    }
    LOG_DEBUG("read over");
    return len;
}

ssize_t st_splice_in(st_netfd_t fd_in, int fd_out, loff_t *off_out, size_t count, int pipes[], unsigned int flags)
{
    ssize_t _count = count;
    while(count > 0){
        ssize_t n = splice(st_netfd_fileno(fd_in), NULL, pipes[1], NULL, count, flags);
        // ssize_t n = splice(st_netfd_fileno(fd_in), NULL, fd_out, off_out, count, flags);
        if (unlikely(n < 0)){
            if (unlikely((errno == EINTR)))
            {
                LOG_INFO("splice errno=EINTR, continue %m");
                continue;
            }

            if (unlikely(!_IO_NOT_READY_ERROR))
            {
                LOG_INFO("splice _IO_NOT_READY_ERROR %m %d %d %d %d %x", st_netfd_fileno(fd_in), fd_out, *off_out, count, flags);
                return -1;
            }

            if (unlikely(st_netfd_poll(fd_in, POLLIN, URSAX_TIMEOUT) < 0))
            {
                LOG_INFO("splice st_netfd_poll wait for socket writable error, %m");
                return -1;
            }
            // LOG_INFO("splice continue, %m");
            continue;
        }else if(unlikely(n == 0))
        {
            LOG_INFO("splice fd sock %d fd file %d offset %d count %d %m", st_netfd_fileno(fd_in), fd_out, *off_out, count);
            return _count - count;
        }
        int ret = splice(pipes[0], NULL, fd_out, off_out, n, flags);
        if(ret != n) {
            LOG_WARN("splice ret != n");
        }
        count -= n;
    }
    return _count;
}

extern uint64_t sendfile_count, st_write_count;
extern uint64_t sendfile_size, st_write_size;
static inline ssize_t try_sendfile_st_write(st_netfd_t stfd, void *buf, size_t nbyte, st_utime_t timeout)
{
    int r;
    struct MMapPair fd_pair = search_mmap(buf);
    if(fd_pair.fd < 0){
        st_write_count++;
        st_write_size += nbyte;
        r = st_write(stfd, buf, nbyte, timeout);
    } else {
        sendfile_count++;
        sendfile_size += nbyte;
        off_t file_offset = fd_pair.offset;
        r = st_sendfile(stfd, fd_pair.fd, &file_offset, nbyte, NULL);
    }
    return r;
}

static inline ssize_t try_splice_recive_st_read(st_netfd_t stfd, void *buf,
        size_t nbyte, st_utime_t timeout)
{
    int r;
    struct MMapPair fd_pair = search_mmap(buf);
    if(fd_pair.fd < 0){
        st_write_count++;
        r = st_read_fully(stfd, buf, nbyte, timeout);
    } else {
        sendfile_count++;
        off_t file_offset = fd_pair.offset;
        r = st_splice_in(stfd, fd_pair.fd, &file_offset, nbyte, fd_pair.pipe,
                SPLICE_F_NONBLOCK|SPLICE_F_MOVE);
    }
    return r;
}

static inline int multy_send_st_writev(st_netfd_t stfd, const struct iovec *iov, int iov_size, uint32_t size, st_utime_t timeout)
{
    int len = 0;
    if(multy_send && size >= splice_line) {
        for(int i = 0; i < iov_size; i++) {
            _put_read_lock(stfd);
            _get_write_lock(stfd);
            int r = try_sendfile_st_write(stfd, iov[i].iov_base, iov[i].iov_len, timeout);
            if(r < 0){
                len = r;
                break;
            }
            len += r;
        }
    }
    else {
        len = st_writev(stfd, iov, iov_size, timeout);
    }
    return len;
}

int multy_receive_st_readv(st_netfd_t stfd, struct iovec **iov, int *iov_size,
        uint32_t size, st_utime_t timeout)
{
    int ret = -1;
    if(unlikely(multy_send && size >= splice_line)) {
        for(int i = 0; i < *iov_size; i++){
            _put_write_lock(stfd);
            _get_read_lock(stfd);
            size_t r = try_splice_recive_st_read(stfd, (*iov)[i].iov_base,
                    (*iov)[i].iov_len, timeout);
            if(r != (*iov)[i].iov_len) {
                ret = -1;
                goto out;
            }
        }
        *iov_size = 0;
        ret = 0;
    }
    else {
        ret = st_readv_resid(stfd, iov, iov_size, URSAX_TIMEOUT);
    }
    out:
    return ret;
}

int operation_sendv_recv(st_netfd_t stfd, void* req, uint32_t req_size, struct iovec *wiov,
                         uint32_t vec_num, uint32_t size, void* resp, uint32_t resp_size)
{
    struct iovec iov[DEFAULT_VECTOR_SIZE];
    iov[0].iov_base = req;
    iov[0].iov_len = req_size;
    memcpy(iov+1, wiov, vec_num * sizeof(struct iovec));
    LOG_DEBUG("%s:start write", __func__);

    int len = multy_send_st_writev(stfd, iov, vec_num + 1, size, URSAX_TIMEOUT);
    if (unlikely(len < (int)(req_size+size)))
    {
        int errno_st_writev = errno;
        log_socket_error(stfd, "writev");
        return -errno_st_writev;
    }
    flush_tcp(stfd);

    LOG_DEBUG("%s:write over, start read", __func__);
    len = st_read_fully(stfd, resp, resp_size, URSAX_TIMEOUT);
    if (unlikely(len < (int)resp_size))
    {
        int errno_st_read_fully = errno;
        log_socket_error(stfd, "read");
        return -errno_st_read_fully;
    }
    LOG_DEBUG("%s:read over", __func__);
    return len;
}

int wait_socket_ready(int fd, bool read, long utime_out)
{
    struct pollfd wfd[1];

    wfd[0].fd     = fd;
    wfd[0].events = read ? POLLIN : POLLOUT;
    wfd[0].revents = 0;

    int res;
    poll:
    res = st_poll(wfd, 1, utime_out);
    if (res == -1) {
        if(errno == EINTR){
            LOG_INFO("st pool errno == EINTR, retry");
            errno = 0;
            goto poll;
        }
        LOG_ERROR("error poll fd return -1");
        return -2;
    }
    if (res == 0) {
        LOG_ERROR("error poll fd timeout");
        return -1;
    }
    return 0;
}

int operation_send2(st_netfd_t stfd, void* part1, uint32_t part1_size,
        void* part2, uint32_t part2_size, int timeout)
{
    struct iovec iov[2];
    iov[0].iov_base = part1;
    iov[0].iov_len = part1_size;
    iov[1].iov_base = part2;
    iov[1].iov_len = part2_size;
    LOG_DEBUG("start write");
    int len = st_writev(stfd, iov, 2, timeout);
    if (unlikely(len < (int)(part1_size + part2_size))) {
        int errno_st_writev = errno;
        log_socket_error(stfd, "writev");
        return -errno_st_writev;
    }
    return 0;
}

int operation_send2_recv_with_timeout(st_netfd_t stfd, void* part1, uint32_t part1_size, void* part2, uint32_t part2_size, void* resp, uint32_t resp_size, int timeout)
{
    int ret = operation_send2(stfd, part1, part1_size, part2, part2_size, timeout);
    if(ret < 0){
        return ret;
    }
    flush_tcp(stfd);

    LOG_DEBUG("write over, start read");
    int len = st_read_fully(stfd, resp, resp_size, timeout);
    if (unlikely(len < (int)resp_size))
    {
        int errno_st_read_fully = errno;
        log_socket_error(stfd, "read");
        return -errno_st_read_fully;
    }
    LOG_DEBUG("read over");
    return len;
}

int operation_send2_recv(st_netfd_t stfd, void* part1, uint32_t part1_size,
        void* part2, uint32_t part2_size, void* resp, uint32_t resp_size)
{
    return operation_send2_recv_with_timeout(stfd, part1, part1_size, part2,
            part2_size, resp, resp_size, URSAX_TIMEOUT);
}

int operation_send2x_recv(st_netfd_t stfd, void* part1, uint32_t part1_size,
        void* part2, uint32_t part2_size, void* resp, uint32_t resp_size, send_cb_t send)
{
    int len = st_write(stfd, part1, part1_size, URSAX_TIMEOUT);
    if (unlikely(len < (int)part1_size)) {
        int errno_st_writev = errno;
        log_socket_error(stfd, "operation_send2x_recv st_write");
        return -errno_st_writev;
    }
    len = send(stfd, part2, part2_size);
    if(unlikely(len < (int)part2_size)) {
        int errno_st_writev = errno;
        log_socket_error(stfd, "operation_send2x_recv send");
        return -errno_st_writev;
    }
    flush_tcp(stfd);

    LOG_DEBUG("write over, start read");
    len = st_read_fully(stfd, resp, resp_size, URSAX_TIMEOUT);
    if (unlikely(len < (int)resp_size)) {
        int errno_st_read_fully = errno;
        log_socket_error(stfd, "read");
        return -errno_st_read_fully;
    }
    LOG_DEBUG("read over");
    return len;
}

void get_peer_addr(st_netfd_t stfd, uint32_t* ip, uint16_t* port)
{
    struct sockaddr_in remote_addr;
    socklen_t addrlen = sizeof(remote_addr);
    getpeername(st_netfd_fileno(stfd), (struct sockaddr*)&remote_addr, &addrlen);
    *ip = ntohl(remote_addr.sin_addr.s_addr);
    *port = ntohs(remote_addr.sin_port);
}

int reply1(st_netfd_t stfd, void* part1, uint32_t part1_size)
{
    int ret = st_write(stfd, part1, part1_size, URSAX_TIMEOUT);
    if (unlikely(ret < (int)part1_size)) {
        int errno_st_write = errno;
        log_socket_error(stfd, "write");
        return -errno_st_write;
    }
    return ret;
}

int reply2(st_netfd_t stfd, void* part1, uint32_t part1_size, void* part2, uint32_t part2_size)
{
    struct iovec iov[2];
    iov[0].iov_base = part1;
    iov[0].iov_len = part1_size;
    iov[1].iov_base = part2;
    iov[1].iov_len = part2_size;
    int ret = st_writev(stfd, iov, 2, URSAX_TIMEOUT);
    if (unlikely(ret < (int)(part1_size + part2_size))) {
        int errno_st_writev = errno;
        log_socket_error(stfd, "writev");
        return -errno_st_writev;
    }
    return ret;
}